package org.ak47.spring.boot.grpc.autoconfigure;


import io.grpc.*;
import lombok.extern.slf4j.Slf4j;
import org.ak47.spring.boot.grpc.GrpcServerProperties;
import org.ak47.spring.boot.grpc.annotations.GrpcService;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.support.AbstractApplicationContext;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Stream;


/**
 * <p> grpc 启动类 </p>
 *
 * @author hexiaohong
 */
@Slf4j
public class GrpcServerRunner implements CommandLineRunner, DisposableBean
{

    private AbstractApplicationContext applicationContext;

    private GrpcServerProperties properties;

    private GlobalServerInterceptorRegistry interceptorRegistry;

    private Server server;

    public GrpcServerRunner(AbstractApplicationContext applicationContext,
                            GrpcServerProperties properties,
                            GlobalServerInterceptorRegistry interceptorRegistry)
    {
        this.applicationContext = applicationContext;
        this.properties = properties;
        this.interceptorRegistry = interceptorRegistry;
    }

    @Override
    public void run(String... args)
        throws Exception
    {
        log.info("Init Grpc Server ......");

        List<ServerInterceptor> globalServerInterceptors = interceptorRegistry.getServerInterceptors();
        Map<String, Object> grpcServices = applicationContext.getBeansWithAnnotation(
            GrpcService.class);
        if (grpcServices.isEmpty())
        {
            log.warn("not found grpc service ,grpc server stop");
            return;
        }

        ServerBuilder<?> serverBuilder = ServerBuilder.forPort(this.properties.getPort());
        grpcServices.forEach((name, service) -> {
            if (!(service instanceof BindableService))
            {
                throw new IllegalArgumentException(
                    service.getClass().getName() + " Is Not Grpc Service, should be implements "
                                                   + BindableService.class.getName());
            }
            List<ServerInterceptor> interceptorList = new ArrayList<>(globalServerInterceptors);
            BindableService bindableService = (BindableService)service;
            GrpcService grpcServiceAnn = applicationContext.findAnnotationOnBean(name,
                GrpcService.class);
            if (grpcServiceAnn != null)
            {
                Stream.of(grpcServiceAnn.interceptors()).forEach(aClass -> {
                    try
                    {
                        ServerInterceptor interceptor = applicationContext.getBeanNamesForType(
                            aClass).length > 0 ? applicationContext.getBean(
                                aClass) : aClass.newInstance();
                        interceptorList.add(interceptor);
                    }
                    catch (Exception e)
                    {
                        throw new BeanCreationException("Failed to create interceptor instance.",
                            e);
                    }
                });
            }
            ServerServiceDefinition serviceDefinition = bindableService.bindService();
            ServerServiceDefinition intercept = ServerInterceptors.intercept(serviceDefinition,
                interceptorList);
            serverBuilder.addService(intercept);
            log.info("{} has bean registered.", bindableService.getClass().getName());
        });
        this.server = serverBuilder.build().start();
        log.info("gRPC Server started, listening on port {}.", server.getPort());
        this.startDaemonAwaitThread();
    }

    private void startDaemonAwaitThread()
    {
        Thread awaitThread = new Thread(() -> {
            try
            {
                GrpcServerRunner.this.server.awaitTermination();
            }
            catch (InterruptedException var2)
            {
                GrpcServerRunner.log.error("gRPC server stopped.", var2);
            }

        });
        awaitThread.setDaemon(false);
        awaitThread.start();
    }

    @Override
    public void destroy()
        throws Exception
    {
        log.info("Shutting down gRPC server ...");
        Optional.ofNullable(this.server).ifPresent(Server::shutdown);
        log.info("gRPC server stopped.");
    }
}
